package cn.itcast

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

case class Person(id:Int,name:String,age:Int)

object CaseClassSchema {
  def main(args: Array[String]): Unit = {
    //1.构建SparkSession
    val spark : SparkSession = SparkSession.builder()
      .appName("CaseClassSchema")
      .master("local[2]")
      .getOrCreate();
    //2.获取SparkContext
    val sc : SparkContext =spark.sparkContext;
    //设置日志打印级别
    sc.setLogLevel("WARN")
    //3.读取文件
    val data: RDD[Array[String]] = sc.textFile("person.txt").map(x=>x.split(" "));
    //4.将RDD与样例类关联
    val personRdd: RDD[Person] = data.map(x=>Person(x(0).toInt,x(1),x(2).toInt))
    //5.获取DF
    //手动导入隐式转换
    import spark.implicits._
    val personDF: DataFrame = personRdd.toDF
    //------------DSL语法操作开始-------------
    //1、显示DataFrame的数据，默认显示20行
    personDF.show()
    //2、显示DataFrame的schema信息
    personDF.printSchema()
    //3、显示DataFrame记录数
    println(personDF.count())
    //4、显示DataFrame的所有字段
    personDF.columns.foreach(println)
    //5、取出DataFrame的第一行记录
    println(personDF.head())
    //6、显示DataFrame中name字段的所有值
    personDF.select("name").show()
    //7、过滤出DataFrame中年龄大于30的记录
    personDF.filter($"age" > 30).show()
    //8、统计DataFrame中年龄大于30的人数
    println(personDF.filter($"age">30).count())
    //9、统计DataFrame中按照年龄进行分组，求每个组的人数
    personDF.groupBy("age").count().show()
    //-----------DSL语法操作结束-------------
    //-----------SQL操作风格开始-------------
    //将DataFrame注册成表
    personDF.createOrReplaceTempView("t_person")
    //传入sql语句，进行操作
    spark.sql("select * from t_person").show()
    spark.sql("select * from t_person where name='zhangsan'").show()
    spark.sql("select * from t_person order by age desc").show()
    //-----------SQL操作风格结束-------------
    //关闭操作
    sc.stop()
    spark.stop()
  }
}
